package com.cike.sparkstudy.core.java;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.VoidFunction;
import scala.Tuple2;

import java.util.Arrays;
import java.util.Iterator;
import java.util.List;

public class TransformationOperation {
    private static final long serialVersionUID = 4603413242796444609L;

    public static void main(String args[]){
        SparkConf conf = new SparkConf().setAppName("TransformationOperation").setMaster("local");

        JavaSparkContext sc = new JavaSparkContext(conf);

        //maptest(sc);
        // flat(sc);
        //flatMap(sc);
        //groupByKey(sc);
        //reduceByKey(sc);
        //sortByKey(sc);
        //join(sc);
    }

    /**
     * 一个集合中所有的数*2
     * @param sc
     */
    public static void maptest(JavaSparkContext sc){
        List<Integer> numbers = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 85, 3, 43,0 ,4);

        JavaRDD<Integer> numbersRDD = sc.parallelize(numbers);

        JavaRDD<Integer> multiplenumbersRDD = numbersRDD.map(new Function<Integer, Integer>() {
            @Override
            public Integer call(Integer v1) throws Exception {
                return v1 * 2;
            }
        });

        multiplenumbersRDD.foreach(new VoidFunction<Integer>() {
            @Override
            public void call(Integer num1) throws Exception {
                System.out.printf(num1+"  ");
            }
        });

        sc.close();

    }

    /**
     * 过滤掉不符合条件的数据（集合中数值为偶数的数据）
     * @param sc
     */
    public static void flat(JavaSparkContext sc){

        List<Integer> numbers = Arrays.asList(1, 2, 3, 4, 45, 3, 6, 7, 8, 10, 49, 0, 434);

        JavaRDD<Integer> numberRDD = sc.parallelize(numbers);

        /**
         * filter算子是过滤掉不符合计算逻辑的数据，符合的返回true且保留下来，不符合则返回false不保修
         */
        JavaRDD<Integer> evenNumberRDD = numberRDD.filter(new Function<Integer, Boolean>() {
            private static final long serialVersionUID = -246390023054245704L;

            @Override
            public Boolean call(Integer v1) throws Exception {
                return v1 % 2 == 0;
            }
        });

        evenNumberRDD.foreach(new VoidFunction<Integer>() {
            private static final long serialVersionUID = 5684650866373972312L;

            @Override
            public void call(Integer v2) throws Exception {
                System.out.println(v2);
            }
        });

        sc.close();
    }

    /**
     * 将行数据按照规定分隔符拆成多个单词
     * @param sc
     */
    public static void flatMap(JavaSparkContext sc){

        List<String> lineList = Arrays.asList("hello you", "hello me", "hello word", "like money");

        JavaRDD<String> linesRDD = sc.parallelize(lineList);

        /**
         * 在这里flatMap会提示把一行数据根据空格切割成多个单词
         * FlatMapFunction第二个参数是返回的数据类型
         * call()方法，返回的类型，不是U，而是Iterable<U>，这里的U也与第二个泛型类型相同
         * flatMap其实就是，接收原始RDD中的每个元素，并进行各种逻辑的计算和处理，返回可以返回多个元素
         * 多个元素，即封装在Iterable集合中，可以使用ArrayList等集合
         * 新的RDD中，即封装了所有的新元素；也就是说，新的RDD的大小一定是 >= 原始RDD的大小
         */
        JavaRDD<String> words = linesRDD.flatMap(new FlatMapFunction<String, String>() {
            private static final long serialVersionUID = -8813632147848744722L;
            /**
             * 比如，在这里传入hello you,那么就会返回Iterable<String>(hello,you)
             * @param s
             * @return
             * @throws Exception
             */
            @Override
            public Iterable<String> call(String s) throws Exception {
                return Arrays.asList(s.split(" "));
            }
        });

        words.foreach(new VoidFunction<String>() {
            private static final long serialVersionUID = 9166285862442789351L;

            @Override
            public void call(String s) throws Exception {
                System.out.println(s);
            }
        });

        sc.close();

    }

    /**
     * 根据键进行分组
     * @param sc
     */
    public static void groupByKey(JavaSparkContext sc){
        //模拟集合
        List<Tuple2<String, Integer>> sorceList = Arrays.asList(
                new Tuple2<String, Integer>("class1", 100),
                new Tuple2<String, Integer>("class2", 50),
                new Tuple2<String, Integer>("class1", 80),
                new Tuple2<String, Integer>("class2", 90),
                new Tuple2<String, Integer>("class1", 77),
                new Tuple2<String, Integer>("class2", 20));

        JavaPairRDD<String, Integer> sorceLists = sc.parallelizePairs(sorceList);

        // 针对scores RDD，执行groupByKey算子，对每个班级的成绩进行分组
        // groupByKey算子，返回的还是JavaPairRDD
        // 但是，JavaPairRDD的第一个泛型类型不变，第二个泛型类型变成Iterable这种集合类型
        // 也就是说，按照了key进行分组，那么每个key可能都会有多个value，此时多个value聚合成了Iterable
        // 那么接下来，我们是不是就可以通过groupedScores这种JavaPairRDD，很方便地处理某个分组内的数据
        JavaPairRDD<String, Iterable<Integer>> groupSorces = sorceLists.groupByKey();

        groupSorces.foreach(new VoidFunction<Tuple2<String, Iterable<Integer>>>() {
            private static final long serialVersionUID = 6141912772349983038L;

            @Override
            public void call(Tuple2<String, Iterable<Integer>> s) throws Exception {
                System.out.println("class1："+s._1);
                Iterator<Integer> ite = s._2.iterator();
                while (ite.hasNext()){
                    System.out.println(ite.next());
                }
                System.out.println("============================");
            }
        });
        sc.close();
    }

    /**
     * 根据键进行聚合
     */
    public static void reduceByKey(JavaSparkContext sc){
        //模拟集合
        List<Tuple2<String, Integer>> sorceList = Arrays.asList(
                new Tuple2<String, Integer>("class1", 100),
                new Tuple2<String, Integer>("class2", 50),
                new Tuple2<String, Integer>("class1", 80),
                new Tuple2<String, Integer>("class2", 90),
                new Tuple2<String, Integer>("class1", 77),
                new Tuple2<String, Integer>("class2", 20));

        JavaPairRDD<String, Integer> sorceLists = sc.parallelizePairs(sorceList);

        /**
         * 针对sorceListsRDD执行reduceByKey算子，
         * reduceByKey算子接受的是Fuction2方法参数，它有三个参数类型
         * 因此对每个key进行reduce，都会依次将第一个、第二个value传入，将值再与第三个value传入
         * 因此此处，会自动定义两个泛型类型，代表call()方法的两个传入参数的类型
         * 第三个泛型类型，代表了每次reduce操作返回的值的类型，默认也是与原始RDD的value类型相同的
         * reduceByKey算法返回的RDD，还是JavaPairRDD<key, value>
         */
        JavaPairRDD<String, Integer> sorces = sorceLists.reduceByKey(new Function2<Integer, Integer, Integer>() {
            private static final long serialVersionUID = -7566506532349125207L;
            @Override
            public Integer call(Integer v1, Integer v2) throws Exception {
                return v1 + v2;
            }
        });

        sorces.foreach(new VoidFunction<Tuple2<String, Integer>>() {
            @Override
            public void call(Tuple2<String, Integer> t) throws Exception {
                System.out.printf(t._1+":"+t._2);
            }
        });

        sc.close();
    }

    /**
     * 将学生分数进行排序
     *
     */
    public static void sortByKey(JavaSparkContext sc){

        //模拟集合
        List<Tuple2<Integer, String>> sorceList = Arrays.asList(
                new Tuple2<Integer, String>(100, "liao"),
                new Tuple2<Integer, String>(90, "李四"),
                new Tuple2<Integer, String>(95, "张三"),
                new Tuple2<Integer, String>(40, "王五"),
                new Tuple2<Integer, String>(60, "小明"),
                new Tuple2<Integer, String>(80, "李磊"));

        JavaPairRDD<Integer, String> sorceLists = sc.parallelizePairs(sorceList);

        /**
         * 对sorceListsRDD 进行排序
         * sortByKey对键进行排序，可以手动指定升序还是降序
         * 返回的还是JavaPairRDD，但是顺序和原始的不一样
         */
        JavaPairRDD<Integer, String> sortedSorce = sorceLists.sortByKey(false);

        sortedSorce.foreach(new VoidFunction<Tuple2<Integer, String>>() {
            @Override
            public void call(Tuple2<Integer, String> t) throws Exception {
                System.out.println(t._1+":"+t._2);
            }
        });
        sc.close();
    }

    /**
     *
     * @param sc
     */
    public static void join(JavaSparkContext sc){
        // 模拟集合
        List<Tuple2<Integer, String>> studentList = Arrays.asList(
                new Tuple2<Integer, String>(1, "leo"),
                new Tuple2<Integer, String>(2, "jack"),
                new Tuple2<Integer, String>(3, "tom"));

        List<Tuple2<Integer, Integer>> scoreList = Arrays.asList(
                new Tuple2<Integer, Integer>(1, 100),
                new Tuple2<Integer, Integer>(2, 90),
                new Tuple2<Integer, Integer>(3, 60));

        // 并行化两个RDD
        JavaPairRDD<Integer, String> students = sc.parallelizePairs(studentList);
        JavaPairRDD<Integer, Integer> scores = sc.parallelizePairs(scoreList);

        /**
         * 对
         *
         */
        JavaPairRDD<Integer, Tuple2<String, Integer>> studentScores = students.join(scores);

        studentScores.foreach(new VoidFunction<Tuple2<Integer, Tuple2<String, Integer>>>() {
            private static final long serialVersionUID = 6496905965551791234L;

            @Override
            public void call(Tuple2<Integer, Tuple2<String, Integer>> t) throws Exception {
                System.out.println("studentID is :"+t._1);
                System.out.println("studentID is :"+t._2._1);
                System.out.println("studentID is :"+t._2._2);
                System.out.println("========================================");

            }
        });


    }

}
